package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.edu.realtime.app.func.DorisSinkFunction;
import com.atguigu.edu.realtime.app.func.MyAsyncFunction;
import com.atguigu.edu.realtime.beans.TrafficPageViewBean;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.DorisUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

//浏览域各来源各访客类别粒度页面浏览各窗口汇总表
//独立访客数,会话总数,浏览页面数,浏览总时长
//DwsTrafficSourceIsNewSessionViewWindow
//session
public class DwsTrafficSourceIsNewViewWindow {
    public static void main(String[] args) {
        //基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);
        //检查点配置
        /*env.enableCheckpointing(5000l);
        env.getCheckpointConfig().setCheckpointTimeout(60000l);
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000l);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(10)));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/edu/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/
        //读取kafka的dwd_traffic_page_log主题
        DataStreamSource<String> jsonStrDs = env.fromSource(MyKafkaUtil.getKafkaSource("dwd_traffic_page_log", "dws_traffic_view_group"),
                WatermarkStrategy.noWatermarks(), "kafkassorce");
        //jsonStrDs.print("3333");
        //过滤数据并转换数据类型
        SingleOutputStreamOperator<JSONObject> jsonObjDs = jsonStrDs.map(JSON::parseObject);
        //对mid分组
        KeyedStream<JSONObject, String> midKeyedStream = jsonObjDs.keyBy(
                value -> value.getJSONObject("common").getString("mid")
        );
        //将数据封装成对象
        SingleOutputStreamOperator<TrafficPageViewBean> orderBeanDs = midKeyedStream.process(
                new ProcessFunction<JSONObject, TrafficPageViewBean>() {


                    private ValueState<String> uvCtState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<String> stateProperties = new ValueStateDescriptor<String>("uvCtState", String.class);
                        stateProperties.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        uvCtState = getRuntimeContext().getState(stateProperties);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, TrafficPageViewBean>.Context ctx, Collector<TrafficPageViewBean> out) throws Exception {
                        JSONObject common = jsonObj.getJSONObject("common");
                        String isNew = common.getString("is_new");
                        String sc = common.getString("sc");
                        //取时间字段
                        Long ts = jsonObj.getLong("ts");
                        String date = DateFormatUtil.toDate(ts);
                        //取状态
                        String value = uvCtState.value();
                        Long uvCt = 0l;
                        //判断状态是否为空
                        if (StringUtils.isEmpty(value)) {
                            //为空则是独立访客,并把时间写入状态
                            uvCtState.update(date);
                            uvCt = 1l;
                        } else {
                            //不为空则判断是否与状态日期相同,不同则是独立房客,更新状态
                            if (!value.equals(date)) {
                                uvCtState.update(date);
                                uvCt = 1l;
                            }
                        }
                        //判断是否为新会话
                        JSONObject page = jsonObj.getJSONObject("page");
                        Long svCt = StringUtils.isEmpty(page.getString("last_page_id")) ? 1l : 0l;
                        out.collect(new TrafficPageViewBean(
                                "",
                                "",
                                sc,
                                "",
                                isNew,
                                "",
                                uvCt,
                                svCt,
                                1l,
                                page.getLong("during_time"),
                                ts
                        ));

                    }
                }
        );
        //orderBeanDs.print("222");
        //设定水位线
        SingleOutputStreamOperator<TrafficPageViewBean> watermarksDs = orderBeanDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TrafficPageViewBean>() {
                                    @Override
                                    public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );

        //分组
        KeyedStream<TrafficPageViewBean, Tuple2<String, String>> keyedStream = watermarksDs.keyBy(
                new KeySelector<TrafficPageViewBean, Tuple2<String, String>>() {

                    @Override
                    public Tuple2<String, String> getKey(TrafficPageViewBean value) throws Exception {
                        return Tuple2.of(value.getIsNew(), value.getSc());
                    }
                }
        );
        //开窗
        WindowedStream<TrafficPageViewBean, Tuple2<String, String>, TimeWindow> windowDs = keyedStream.window(
                TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))
        );
        //聚合计算
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDs = windowDs.reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                       // System.out.println(value1.toString());
                        return value1;
                    }
                },
                new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple2<String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple2<String, String> stringStringTuple2, ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple2<String, String>, TimeWindow>.Context context, Iterable<TrafficPageViewBean> elements, Collector<TrafficPageViewBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        String date = DateFormatUtil.toDate(context.window().getStart());

                        TrafficPageViewBean viewBean = elements.iterator().next();
                        viewBean.setStt(stt);
                        viewBean.setEdt(edt);
                        viewBean.setCurDate(date);

                        out.collect(viewBean);
                    }
                }
        );
        //维度关联
        SingleOutputStreamOperator<TrafficPageViewBean> joinDs = AsyncDataStream.unorderedWait(
                reduceDs,
                new MyAsyncFunction<TrafficPageViewBean>("dim_base_source") {
                    @Override
                    public void join(JSONObject jsonObject, TrafficPageViewBean orderBean) {
                        orderBean.setScName(jsonObject.getString("source_site"));
                    }

                    @Override
                    public String getKey(TrafficPageViewBean orderBean) {
                        return orderBean.getSc();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //写入到doris中
        joinDs.print(">>>");
        joinDs.addSink(new DorisSinkFunction<TrafficPageViewBean>("dws_traffic_sc_is_new_page_view_window"));



        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
